package com.inspinia.message.common.util;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.inspinia.base.cache.RedisMQ;
import com.inspinia.base.mq.rabbitmq.config.RabbitConfig;
import com.inspinia.base.mq.rabbitmq.service.RabbitMqConsumerRegister;
import com.inspinia.base.mq.rabbitmq.service.RabbitMqMessageReceiver;
import com.inspinia.base.mq.rabbitmq.service.RabbitMqMessageSender;
import com.inspinia.base.util.JsonUtils;
import com.inspinia.message.common.exception.MessageException;
import com.inspinia.message.common.model.NoticeConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import redis.clients.jedis.JedisPubSub;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Collection;
import java.util.concurrent.*;

/**
 * 推送消息工具类
 * 使用mq做消息持久化
 * @author Veblen
 */
@Component
public class MessagePushUtil {

    public static final String NOTIFY_CHANNEL = "notify";
    private static final String queueName = RabbitConfig.QUEUE;

    private static Logger logger = LoggerFactory.getLogger(MessagePushUtil.class);

    @Resource
    RabbitMqConsumerRegister mqConsumerRegister;

    /**
     * 消息推送工具类初始化完成监听消息
     */
    @PostConstruct
    public void init() {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("messagePush-pool-%d").build();
        ExecutorService pool = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

        switch (NoticeConfig.NOTICE_MQ_TYPE) {
            case REDIS:
                pool.execute(() -> RedisMQ.subscribe(new JedisPubSub() {
                    @Override
                    public void onMessage(String channel, String message) {
                        Long userId = getUserIdFromJsonStr(message);
                        WebSocketServerHandler.sendMessage(userId, message.replace(userId + "-", ""));
                    }
                }, NOTIFY_CHANNEL));
                break;
            case RABBITMQ:
                //启动一个线程异步发送
                pool.execute(() -> {
                    RabbitMqMessageReceiver consumer = new RabbitMqMessageReceiver(mqConsumerRegister, queueName, message -> {
                        try {
                            Long userId = getUserIdFromJsonStr(message);
                            WebSocketServerHandler.sendMessage(userId, message.replace(userId + "-", ""));
                            return true;
                        }catch (Exception e){
                            throw new MessageException("消息推送异常");
                        }
                    });
                    consumer.normalReceive();
                });
                break;
            case NONE:
                break;
            default:
                break;
        }
    }

    private static Long getUserIdFromJsonStr(String jsonStr) {
        return Long.valueOf(jsonStr.split("-")[0]);
    }

    private static String genNoticeJsonStr(Long userId, Object msg) {
        String json = JsonUtils.toJson(msg);
        return userId + "-" + json;
    }

    public static void pushJsonMessage(Long userId, Object msg) {
        String jsonMsg = JsonUtils.toJson(msg);
        logger.info("推送消息到:{}",userId);
        switch (NoticeConfig.NOTICE_MQ_TYPE) {
            case NONE:
                WebSocketServerHandler.sendMessage(userId,jsonMsg);
                logger.info("普通推送，消息内容:{}",jsonMsg);
                break;
            case REDIS:
                RedisMQ.publish(queueName, genNoticeJsonStr(userId, msg));
                logger.info("Redis推送，消息内容:{}",jsonMsg);
                break;
            case RABBITMQ:
                RabbitMqMessageSender.sendMsg(queueName,genNoticeJsonStr(userId, msg));
                logger.info("RabbitMq推送，消息内容:{}",jsonMsg);
                break;
            default:
                break;

        }
    }


    public static void pushJsonMessage(Collection<Long> userIds, Object msg) {
        for (Long userId : userIds) {
            pushJsonMessage(userId, msg);
        }
    }


}
